package com.no_spring.hellokafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;

//kafka消费者，helloworld版本
public class C1 {
    public static void main(String[] args) {
        //消费者三个属性必须指定：broker:地址清单，key和value的序列化器
        Properties props = new Properties();
        props.put("bootstrap.servers","127.0.0.1:9092");
        props.put("key.deserializer", StringDeserializer.class);
        props.put("value.deserializer",StringDeserializer.class);

        //消费者群组
        props.put(ConsumerConfig.GROUP_ID_CONFIG,"group01");
        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(props);

        try{
            //消费者订阅主题,可以订阅多个
            consumer.subscribe(Collections.singletonList("topic1"));
            while(true){
                //拉去（新版）
                ConsumerRecords<String,String> records = consumer.poll(500);
                for(ConsumerRecord<String,String> record: records){
                    System.out.println(String.format("topic:%s,分区：%d,偏移量：%d,"+
                            "key:%s,value:%s",record.topic(),record.partition(),
                            record.offset(),record.key(),record.value()));
                }
            }

        }finally {
            consumer.close();
        }
    }
}
